package com.example.demoapachekafka.service.impl;


import com.alibaba.fastjson2.JSON;
import com.example.demoapachekafka.model.*;
import com.example.demoapachekafka.producer.BookInfoToMessageTransfer;
import com.example.demoapachekafka.producer.KafkaMessageProducer;
import com.example.demoapachekafka.service.IExperiment;
import jakarta.annotation.Resource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.stereotype.Component;

import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @Description 下单实验
 * @Date 2021/1/4 10:50 上午
 * @Created by qinshu
 */
@Component
public class BookExperiment implements IExperiment {

    private static Log log = LogFactory.getLog(BookExperiment.class);

    @Resource
    private GoodsSnapshotController goodsSnapshotController;

    @Resource
    private KafkaMessageProducer producer;

    private ExecutorService es = Executors.newFixedThreadPool(10);

    @Override
    public void test() {
        generateOrders();
    }

    // 模拟并发下单
    public void generateOrders() {
        for (int i=1; i < 1000; i++) {
            es.submit(() -> {
                book();
            });
        }
    }

    Random random = new Random(System.currentTimeMillis());

    public BaseResult book() {
        BookInfo bookInfo = new BookInfo();
        OrderDto orderDto = new OrderDto();

        Long shopId = 654321L + random.nextInt(10000);
        Long userId = 1234L + random.nextInt(1000);
        Long goodsId = 5678L + random.nextInt(4000);
        orderDto.setShopId(shopId);
        orderDto.setUserId(userId);
        orderDto.setDeliveryType(DeliveryType.express);
        orderDto.setIsCodPay(false);
        bookInfo.setOrderDto(orderDto);

        GoodsInfo goods = new GoodsInfo();
        goods.setGoodsId(goodsId);
        goods.setShopId(shopId);
        goods.setTitle("认养一头牛");
        goods.setDesc("2箱*250g");
        bookInfo.setGoods(goods);

        BaseResult bookResult = goodsSnapshotController.save(bookInfo);
        log.info("下单结果:" + JSON.toJSONString(bookResult));

        // 下单成功后发送消息
        producer.sendAsync(
                BookInfoToMessageTransfer.transfer(bookInfo),
                (metadata, exception) -> callback(bookInfo, metadata, exception));

        return bookResult;
    }

    // 消息发送后的回调函数
    private void callback(BookInfo bookInfo, RecordMetadata metadata, Exception ex) {
        if (metadata != null) {
            log.info("发送订单消息:" + bookInfo.getOrderDto().getOrderNo() + " 偏移量: " + metadata.offset() + " 主题: " + metadata.topic());
        } else {
            log.error("发送订单消息失败: " + ex.getMessage(), ex);
        }
    }
}
